/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.tools;

import cn.ac.iie.di.ban.commons.RuntimeEnv;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.FailureMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Level;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.log4j.Logger;

/**
 *
 * @author xiaoliang
 */
public class RocketMqHelper {

    private static Logger logger = Logger.getLogger(RocketMqHelper.class.getName());
    private Object consumerLock = new Object();
    private String MQName;
    private REReceiveSession consumer;
    private BlockingQueue<List<Map<String,Object>>> dataCache;
    private Map<String,REFieldType> receiveColMap = new HashMap<>();
    private String group;
    
    public RocketMqHelper(String MQname,BlockingQueue<List<Map<String,Object>>> dataCache,String group) {
        this.MQName = MQname;
        this.dataCache = dataCache;
        this.group = group;
    }

    public void init() {
    }

    public void registerMsgHandler() throws Exception {
        synchronized (consumerLock) {
            if (consumer == null) {
                REConnection conn = new REConnection(RuntimeEnv.getParamAsString(RuntimeEnv.NAME_SERVER_LIST));
                REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(MQName);
                logger.info("rocketmqpull from " + RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM));
                if (RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM).equals("last")) {
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
                } else if (RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM).equals("first")) {
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_FIRST_OFFSET);
                } else if (RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM).equals("time")) {
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_TIMESTAMP);
                    String ftime = RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM_Time);
                    try {
//                        long ft = Integer.valueOf(ftime);
                        Long time = time2stamp(ftime);
                        logger.info("pull data from mq and time is " + ftime);
                        builder.setConsumeTimestamp(time);
                    } catch (Exception E) {
                        logger.error(E.getMessage(), E);
                        long ft = System.currentTimeMillis() - 1800000;
                        builder.setConsumeTimestamp(ft);
                        logger.info("not get time from conf_rocketMQPullFromTime and is " + RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM_Time) + "and pull data from last 30 mints ago and time is " + ftime);
                    }

                } else {
                    logger.warn("rocketmqpull from   last unknow is " + RuntimeEnv.getParamAsString(RuntimeEnv.ROCKETMQ_PULL_FROM));
                    builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
                }
                Integer threadNum = 2;
                String threadString = (String) RuntimeEnv.getParam(RuntimeEnv.THREAD_NUM);
                if (!"".equals(threadString) && threadString != null) {
                    threadNum = Integer.parseInt(threadString);
                }
                builder.setConsumeThreadNum(threadNum);
                logger.info("ConsumeThreadNum is " + threadNum);
                builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>() {
                    @Override
                    public boolean handle(REMessage message) {
                        disposeMessages(message);
                        return true;
                    }
                });
                builder.setFailureHandler(new REAbstractReceiveMessageHandler<FailureMessage>() {
                    @Override
                    public boolean handle(FailureMessage msg) {
                        logger.error("formatter fuilure ! size = " + msg.getMsg().length, msg.getEx());
                        return true;
                    }
                });

                builder.setGroupName("iie_di_" + RuntimeEnv.getParam(RuntimeEnv.REGION) + "_" + group);
                consumer = (REReceiveSession) builder.build();
                logger.info("init consumer group is " +group+ " successfully");
            }
        }
        logger.info("register handler of mq " + MQName + " successfully");
    }

    private void disposeMessages(REMessage msg) {
        try {
            logger.info(Thread.currentThread().getId() + " receives message " + msg.data.size() + " at " + System.nanoTime() + " docschma " + msg.docSchemaName + " from " +MQName);
            try {
                if (receiveColMap.isEmpty()) {
//                    HashMap<String, REFieldType> newtable = new HashMap<String, REFieldType>();
                    for (int i = 0; i < msg.colNames.length; i++) {
                        receiveColMap.put(msg.colNames[i], msg.types[i]);//mq中的字段。和字段类型
                    }
//                    for (Map.Entry<String, String> entry : RuntimeEnv.TABLE_RC_COLUMNS.get(tableSe2DBRule.getRuleName()).entrySet()) {
//                        String rname = entry.getKey();
//                        if (newtable.get(rname) == null) {
//                            logger.debug("configColumns compare rocketmqColumns and roc is null " + rname);
//                        }
//                    }
//                    logger.info("docSchemaName " + msg.docSchemaName + " desc HashMap " + newtable);
//                    ROCKETMQ_DOC_DESC.put(tabname, newtable);
                }
                if (msg.data.size() < 1) {
                    logger.warn("doc set is empty");
                    return;
                }
            } catch (Exception ex) {
                logger.error("parse docs unsuccessfully for " + ex.getMessage(), ex);
                return;
            } finally {

            }
            List<Map<String,Object>> recordSet = new ArrayList<Map<String,Object>>();
            int colLen = msg.colNames.length;
            for (Object[] doc : msg.data) {
                HashMap<String, Object> hm = new HashMap<String, Object>();
                for (int i = 0; i < colLen; ++i) {
                    if (doc[i] == null) {
                        hm.put(msg.colNames[i], null);
                        continue;
                    }
                    hm.put(msg.colNames[i], doc[i]);
                }
                recordSet.add(hm);
            }
            int bufferTime = 0;
            while (true) {
                logger.debug("dataCache size:" + dataCache.size());
                if (!dataCache.offer(recordSet)) {
                    if (bufferTime++ > 20) {
                        logger.info("from topic:"+MQName + "to table buffer fail,will retry ...");
                        bufferTime = 0;
                    }
                    try {
                        Thread.sleep(200);
                    } catch (Exception ex) {
                    }
                } else {
                    logger.info("analysisSize " + recordSet.size() + "receMsgSize " + msg.data.size());
                    break;
                }
            }
        } catch (Exception ex) {
            logger.fatal("unknown exception " + ex.getMessage(), ex);
        }
    }

    public static Long time2stamp(String time) throws ParseException {
        String regEx = "^\\d{14}$";
        Pattern pattern = Pattern.compile(regEx);
        Matcher matcher = pattern.matcher(time);
        boolean rs = matcher.matches();
        if (rs == false) {
            throw new RuntimeException("input time error!");
        }
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
        Date date = sdf.parse(time);
        return date.getTime() / 1000;
    }
    /*
    push 
    */
    
    
    public void starConsumer(){
        try {
            consumer.start();
        } catch (REConnectionException ex) {
             try {
                consumer.start();
            } catch (Exception e) {
                logger.error(e.getMessage(),e);
            }
        }
    }
    
    public void stopConsumer(){
        try {
            consumer.shutdown();
        } catch (REConnectionException ex) {
            try {
                consumer.shutdown();
            } catch (Exception e) {
            }
        }
    }
}
